feat(realtime): support external offsets and shredding-aware reads - #269

Could this skip an earlier prepared segment?\n\nFor example, if P0 contains offset 10 and P1 contains offset 30, committing only P1 advances the watermark to 31. P0 may then be reclaimed and later treated as already committed, although its files were never published.\n\nI think row offsets can remain sparse, but commit progress should still be contiguous.

Good point. This can happen if the caller passes only P1, but we consider that invalid input to CommitWithProgress.

The range here describes the envelope of actual row offsets, which may be sparse. Therefore, the committer cannot distinguish a legitimate offset gap from an omitted prepared segment, and requiring numeric continuity would reject valid sparse offsets.

Streaming engines generally enforce this at the checkpoint coordinator layer. For example, Flink’s checkpoint-subsuming contract requires a later completed checkpoint to include all artifacts from earlier uncommitted checkpoints. The sink coordinator retains prepared-but-uncommitted committables and submits the complete pending prefix when committing a later checkpoint. Low-level offset APIs such as Kafka also trust the caller to commit the next valid consumption position rather than validating that every preceding record was processed.

We will keep sparse ranges supported and document that the upstream coordinator must provide all earlier prepared-but-uncommitted progress for each partition-bucket. Omitting an earlier prepared segment violates the API contract and may cause its unpublished data to be treated as committed and reclaimed.

Could we validate the raw PK commit readers before MOR?

This path no longer checks segment->GetRowCount(). If a custom store misses or duplicates a mutation, deduplication may hide it and incomplete data could be committed.

We should at least check the raw row count and validate that _REALTIME_OFFSET is non-null, unique, and within the sealed range. The same offset validation may also be useful for append.

Agreed that we can compare the total number of rows emitted by the raw commit readers with segment->GetRowCount(), consistent with the append path.

While I don’t think core should revalidate _REALTIME_OFFSET uniqueness or range here. RealtimeStore::CreateCommitReaders already requires implementations to expose every sealed row exactly once, similar to how core trusts other plugin (e.g., file-format readers) to return correct field values. Offset uniqueness would also require additional state because PK readers are sorted by key and sequence rather than offset.

Even comprehensive offset validation cannot prove that the plugin returned the correct row contents or associated each offset with the correct mutation. Extending validation in this direction would effectively make core revalidate the plugin’s entire output, which is inconsistent with the trust boundary used for other plugin. A custom store violating this contract should be treated as a plugin bug.

This boundary makes sense. The raw row-count check before MOR should be sufficient here.

I agree that validating the raw row count would provide useful defense in depth. However, there is no simple reliable check at this layer with the current streaming reader contract.

Unlike append, PK commit readers are multiple one-shot streams consumed through MOR. The row count after MOR is not comparable because deduplication may reduce it. Checking before MOR would require wrapping every raw reader with shared counting state, detecting completion across all reader EOFs, and propagating a mismatch before the rolling file writer completes. Pre-reading would instead require buffering or reopening the streams.

Since CreateCommitReaders already requires plugins to expose every sealed row exactly once, I will add a TODO/follow-up for cardinality validation.